//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// insert_executor.cpp
//
// Identification: src/execution/insert_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include <memory>

#include "execution/executors/insert_executor.h"

namespace bustub {

InsertExecutor::InsertExecutor(ExecutorContext *exec_ctx, const InsertPlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx), plan_(plan), child_executor_(std::move(child_executor)) {}

void InsertExecutor::Init() {
  child_executor_->Init();
  Catalog *catalog = exec_ctx_->GetCatalog();
  table_info_ = catalog->GetTable(plan_->table_oid_);
  inserted_ = false;
}

auto InsertExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (inserted_) {
    return false;
  }

  Catalog *catalog = exec_ctx_->GetCatalog();
  std::vector<IndexInfo *> indexinfos = catalog->GetTableIndexes(table_info_->name_);

  int value_cnt = 0;
  Tuple ch_tuple;
  RID ch_rid;
  while (child_executor_->Next(&ch_tuple, &ch_rid)) {
    TupleMeta tuple_meta{INVALID_TXN_ID, INVALID_TXN_ID, false};
    RID inserted_rid = table_info_->table_->InsertTuple(tuple_meta, ch_tuple).value();
    ++value_cnt;
    for (IndexInfo *key_info : indexinfos) {
      auto index_meta = key_info->index_->GetMetadata();
      key_info->index_->InsertEntry(
          ch_tuple.KeyFromTuple(table_info_->schema_, *index_meta->GetKeySchema(), index_meta->GetKeyAttrs()),
          inserted_rid, nullptr);
    }
  }

  *tuple = Tuple(std::vector<Value>(GetOutputSchema().GetColumnCount(), Value(TypeId::INTEGER, value_cnt)),
                 &GetOutputSchema());

  inserted_ = true;
  return true;
}

}  // namespace bustub
